#include <stdint.h>
#include <iostream>
#include <sstream>

#include "raptor-lite/raptor-lite.h"

class TcpServerTest : public raptor::AcceptorHandler,
                      public raptor::MessageHandler,
                      public raptor::HeartbeatHandler,
                      public raptor::EndpointClosedHandler {
public:
    TcpServerTest(/* args */);
    ~TcpServerTest();

    void OnAccept(raptor::Endpoint ep, raptor::Property &settings) {
        settings({{raptor::config::kSocketRecvTimeoutMs, 5000},
                  {raptor::config::kSocketSendTimeoutMs, 5000}});
        std::stringstream ss;
        ss << "\nOnAccept:" << ep.PeerString() << std::endl;
        ss << "  fd: " << ep.SocketFd() << std::endl;
        ss << "  RemoteIp: " << ep.RemoteIp() << std::endl;
        ss << "  RemotePort: " << ep.RemotePort() << std::endl;
        ss << "  LocalIp: " << ep.LocalIp() << std::endl;
        ss << "  LocalPort: " << ep.LocalPort() << std::endl;
        ss << "  ConnectionId: " << ep.ConnectionId() << std::endl;
        log_debug("%s", ss.str().c_str());
        container->AttachEndpoint(ep);
    }

    void OnMessage(raptor::Endpoint ep, const raptor::Slice &msg) {
        std::stringstream ss;
        ss << "OnMessage: peer:" << ep.PeerString() << " msg:" << msg.ToString() << std::endl;

        log_debug("%s", ss.str().c_str());
        std::string str = "Reply:" + msg.ToString();
        ep.SendMsg(str);
    }

    // heart-beat event
    void OnHeartbeat(raptor::Endpoint ep) {
        static int64_t current = 0;
        static int i           = 0;
        if (current == 0) {
            current = raptor::GetCurrentMilliseconds();
            log_debug("OnHeartbeat:  fd = %lld", ep.SocketFd());
        } else {
            int64_t now = raptor::GetCurrentMilliseconds();
            log_debug("OnHeartbeat:  fd = %lld, interval = %lld", ep.SocketFd(), now - current);
            current = now;
        }
        ep.SendMsg("This is a heart-beat message! index:" + std::to_string(i++));
    }

    // millseconds
    size_t GetHeartbeatInterval() { return 15 * 1000; }

    void OnClosed(raptor::Endpoint ep, raptor_error desc) {
        std::stringstream ss;
        ss << "\nOnClosed:" << ep.PeerString() << std::endl;
        ss << "  fd: " << ep.SocketFd() << std::endl;
        ss << "  RemoteIp: " << ep.RemoteIp() << std::endl;
        ss << "  RemotePort: " << ep.RemotePort() << std::endl;
        ss << "  LocalIp: " << ep.LocalIp() << std::endl;
        ss << "  LocalPort: " << ep.LocalPort() << std::endl;
        ss << "  ConnectionId: " << ep.ConnectionId() << std::endl;
        if (desc != RAPTOR_ERROR_NONE) {
            ss << "  desc: " << desc->ToString() << std::endl;
        }

        log_debug("%s", ss.str().c_str());
    }

    void init() {
        engine = RaptorCreateEngine();

        raptor::Property p{
            {raptor::config::kAcceptorHandler, static_cast<AcceptorHandler *>(this)},
            {raptor::config::kMessageHandler, static_cast<MessageHandler *>(this)},
            {raptor::config::kHeartbeatHandler, static_cast<HeartbeatHandler *>(this)},
            {raptor::config::kEndpointClosedHandler, static_cast<EndpointClosedHandler *>(this)}};

        raptor_error err = engine->CreateContainer(p, &container);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create container: %s", err->ToString().c_str());
            abort();
        }
        err = engine->CreateAcceptor(p, &acceptor);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to create acceptor: %s", err->ToString().c_str());
            abort();
        }
    }

    void start_and_listening(const std::string &addr) {
        raptor_error err = acceptor->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to Start: %s", err->ToString().c_str());
            return;
        }
        err = container->Start();
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to Start: %s", err->ToString().c_str());
            return;
        }
        err = acceptor->AddListening(addr);
        if (err != RAPTOR_ERROR_NONE) {
            log_error("Failed to Listening: %s", err->ToString().c_str());
            return;
        }
    }

    void stop() {
        acceptor->Shutdown();
        container->Shutdown();
    }

private:
    RaptorEngine engine;
    raptor::Container *container = nullptr;
    raptor::Acceptor *acceptor   = nullptr;
};

TcpServerTest::TcpServerTest(/* args */) {
    // Do not call the raptor::Create function here, because the current
    // class (inherited from XXHandler) has not been constructed.
}

TcpServerTest::~TcpServerTest() {
    engine->DestroyAcceptor(acceptor);
    engine->DestroyContainer(container);
}

int main() {
    printf(" ---- prepare start server ---- \n");
    RaptorGlobalStartup();
    TcpServerTest server;
    server.init();
    server.start_and_listening("localhost:50051");
    log_error("Press any key to quit");
    getchar();
    server.stop();
    RaptorGlobalCleanup();
    return 0;
}
